package com.example.wxg.function;

import org.apache.flink.api.common.accumulators.IntCounter;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

/**
 * @author void
 * @date 2022/1/5 15:12
 * @desc 自定义继承RichFlatMapFunction的分词器，完成分词和累加功能
 */
public class WordSplitRichFlatMap extends RichFlatMapFunction<String, String> {
    
    private int limit;
    
    //累加器
    private IntCounter intCounter = new IntCounter(0);

    public WordSplitRichFlatMap(int limit) {
        this.limit = limit;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        //注册累加器
        getRuntimeContext().addAccumulator("int-counter", this.intCounter);
    }

    @Override
    public void flatMap(String s, Collector<String> collector) throws Exception {
        //执行过程调用累加器
        this.intCounter.add(1);
        
        if(s.length()>limit){
            for(String word : s.split(" ")){
                collector.collect(word);
            }
        }
    }
}
